#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "../MQcommon/logger.hpp"
#include "../MQcommon/helper.hpp"
#include "../MQcommon/mq_msg.pb.h"
#include "../MQcommon/mq_proto.pb.h"
#include "../MQcommon/threadpool.hpp"
#include "mq_consumer.hpp"
#include "mq_host.hpp"
#include "mq_route.hpp"
//信道在逻辑上，也是一个消费者，它将消息从队列当中取出，在将消息发送给客户端
//从大体上看，客户端是消费者，消费服务器存储的信息，但客户端仅仅只是接收并处理来自服务器的信息，服务器上的信息并不是直接被客户端拿走的，中间还有一个信道
//所以实际上，客户端并不是直接消费者，而是间接消费者，信道才是直接消费者，直接得到服务器存储的消息，直接从队列中拿取消息
//一个信道负责一个客户端的通信
namespace bitmq{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;
    using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr = std::shared_ptr<queueUnBindRequest>;
    using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;
    using basicConsumeRequestPtr = std::shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;
    class Channel{
        public:
        using ptr = std::shared_ptr<Channel>;
        Channel(const std::string &id, 
            const VirtualHost::ptr &host, 
            const ConsumerManager::ptr &cmp, 
            const ProtobufCodecPtr &codec, 
            const muduo::net::TcpConnectionPtr &conn,
            const Threadpool::ptr &pool):
            _cid(id),
            _conn(conn),
            _codec(codec),
            _cmp(cmp),
            _host(host),
            _pool(pool){
            DLOG("new Channel: %p", this);
        }
        ~Channel() {
            if (_consumer.get() != nullptr) {
                _cmp->removeConsumer(_consumer->tag, _consumer->qname);
            }
            DLOG("del Channel: %p", this);
        }
        //交换机的创建与删除
        void declareExchange(const declareExchangeRequestPtr &req){
            bool ret = _host->declareExchange(req->exchange_name(),req->exchange_type(),req->durable(),req->auto_delete(),req->args());
            basicResponse(ret,req->rid(),req->cid());
            DLOG("通过信道创建交换机[%s]成功!",req->exchange_name().c_str());
            return ;
        }
        void deleteExchange(const deleteExchangeRequestPtr &req){
            _host->deleteExchange(req->exchange_name());
            return basicResponse(true, req->rid(), req->cid());
        }
        //队列的创建与删除
        void declareQueue(const declareQueueRequestPtr &req){
            bool ret=_host->declareQueue(req->queue_name(),req->durable(),req->exclusive(),req->auto_delete(),req->args());
            if(ret == true){
                _cmp->initQueueConsumer(req->queue_name());
            }
            
            basicResponse(ret,req->rid(),req->cid());
            if(ret == true) DLOG("通过信道创建队列[%s]成功!",req->queue_name().c_str());
            return ;

        }
        void deleteQueue(const deleteQueueRequestPtr &req){
            _cmp->destroyQueueConsumer(req->queue_name());
            _host->deleteQueue(req->queue_name());
             basicResponse(true,req->rid(),req->cid());
             DLOG("通过信道删除队列[%s]成功!",req->queue_name().c_str())
        }
        //队列的绑定与解除绑定
        void queueBind(const queueBindRequestPtr &req){
            bool ret = _host->bind(req->exchange_name(),req->queue_name(),req->binding_key());
            basicResponse(ret,req->rid(),req->cid());
            DLOG("通过信道绑定队列[%s]成功!",req->queue_name().c_str())
        }
        void queueUnBind(const queueUnBindRequestPtr &req){
            _host->unBind(req->exchange_name(),req->queue_name());
            return basicResponse(true,req->rid(),req->cid());
            DLOG("通过信道解绑队列[%s]成功!",req->queue_name().c_str())
        }
        //消息的发布
        void basicPublish(const basicPublishRequestPtr &req){
             //先发送到交换机，再由交换机决定发送到哪个队列，送到队列后，队列此时进行消息推送
             //先判断交换机存在不存在
             auto ex=_host->selectExchange(req->exchange_name());
             if(ex.get() == nullptr){
                DLOG("指定的交换机[%s]不存在！",req->exchange_name().c_str());
                return basicResponse(false,req->rid(),req->cid());
             }
             //根据交换机信息找到对应的绑定信息
             MsgQueueBindingMap mqb=_host->exchangeBindings(req->exchange_name());
             if(mqb.size()== 0){
                DLOG("该交换机[%s]未绑定任何队列!",req->exchange_name().c_str());
                return basicResponse(false,req->rid(),req->cid());
             }
             //找到队列绑定信息后，将每个绑定信息中的r_key进行比对，从而判断向哪些队列中添加消息
             BasicProperties* bp=nullptr;
             std::string routing_key;
             if(req->mutable_properties()){
                bp=req->mutable_properties();
                routing_key=req->properties().routing_key();
             }
             for( auto& x: mqb){
                if(Router::route(ex->type,routing_key,x.second->binding_key)){
                   //如果通过，则将放入该消息放入队列中
                   bool ret=_host->basicPublish(x.first,bp,req->body());
                   if(ret == true){
                    DLOG("消息[%s]已经成功放入队列[%s]",req->body().c_str(),x.first.c_str());
                   }
                   auto task = std::bind(&Channel::consume, this, x.first);
                   _pool->push(task);//消息进入队列后，就需要被消费者处理7
                }
             }
              return basicResponse(false,req->rid(),req->cid());
             //队列中有了消息，就要寻找消费者消费消息
             //这个消费的过程，交给线程池去处理
        }
        //消息的回复
        void basicAck(const basicAckRequestPtr &req){
            _host->basicAck(req->queue_name(),req->message_id());
            basicResponse(true,req->rid(),req->cid());
            DLOG("消息回复成功!",req->queue_name().c_str())
        }
        //订阅队列
        void basicConsume(const basicConsumeRequestPtr &req){
             //先检查队列是否存在
             bool ret=_host->existsQueue(req->queue_name());
             if(ret == false){
                DLOG("订阅的队列[%s]不存在！",req->queue_name().c_str());
                return basicResponse(false,req->rid(),req->cid());
             }
             auto cb=std::bind(&Channel::callBack,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3);
             //向队列的消费者管理者句柄中插入消费者
             _consumer = _cmp->createConsumer(req->consumer_tag(), req->queue_name(), req->auto_ack(), cb);
             basicResponse(true, req->rid(), req->cid());
             DLOG("订阅队列[%s]成功!",req->queue_name().c_str())
        }
        //取消订阅
        void basicCancel(const basicCancelRequestPtr &req){
            bool ret=_host->existsQueue(req->queue_name());
            if(ret == false){
               DLOG("取消订阅的队列[%s]不存在！",req->queue_name().c_str());
               return basicResponse(false,req->rid(),req->cid());
            }
            _cmp->removeConsumer(req->consumer_tag(),req->queue_name());
            basicResponse(true,req->rid(),req->cid());
            DLOG("通过信道取消队列[%s]的订阅！",req->queue_name());
            return ;
        }
        private:
        void consume(std::string& queue_name){
             //取出消息
             MessagePtr msg=_host->basicConsume(queue_name);
             if(msg.get() == nullptr){
                DLOG("暂时没有消息可以处理！");
             }
             //取出消费者
             Consumer::ptr cer=_cmp->chooseConsumer(queue_name);
             if(cer.get() == nullptr){
                DLOG("未在指定队列[%s]中找到消费者!");
                return ;
             }
             cer->callback(cer->tag, msg->mutable_payload()->mutable_properties(), msg->payload().body());
             if(cer->auto_ack){
                _host->basicAck(queue_name,msg->payload().properties().id());
             }


        }
        //消费者的回调函数
        void callBack(const std::string& tag, const BasicProperties *bp, const std::string& body){
            basicConsumeResponse resp;
            resp.set_cid(_cid);
            resp.set_body(body);
            resp.set_consumer_tag(tag);
            if (bp) {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_deliver_mode(bp->deliver_mode());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn, resp);
        }
        void basicResponse(bool ok, const std::string &rid, const std::string &cid) {
            basicCommonResponse resp;
            resp.set_rid(rid);
            resp.set_cid(cid);
            resp.set_ok(ok);
            _codec->send(_conn, resp);
        }
        private:
            std::string _cid;
            Consumer::ptr _consumer;//消费者
            muduo::net::TcpConnectionPtr _conn;//连接管理者
            ProtobufCodecPtr _codec;//协议解释器
            ConsumerManager::ptr _cmp;//消费者管理句柄，可以管理与当前信道关联的所有队列的消费者
            VirtualHost::ptr _host;//虚拟机
            Threadpool::ptr _pool;//异步线程池
    };
    class ChannelManager {
    public:
      using ptr=std::shared_ptr<ChannelManager>;
      ChannelManager(){}
      bool openChannel(const std::string &id, const VirtualHost::ptr &host, 
        const ConsumerManager::ptr &cmp, const ProtobufCodecPtr &codec, 
        const muduo::net::TcpConnectionPtr &conn,const Threadpool::ptr &pool){
            std::unique_lock<std::mutex> lock(_mutex);
           //判断该信道是否已经存在
           auto it = _channel_map.find(id);
           if(it != _channel_map.end()){
              DLOG("要打开的信道[%s]已经存在",id.c_str());
              return true;
           }
           Channel::ptr channel=std::make_shared<Channel>(id,host,cmp,codec,conn,pool);
           auto reslut = _channel_map.insert(std::make_pair(id,channel));
           if(reslut.second){
            DLOG("插入信道[%s]成功",id.c_str());
           }
           return true;
        }
        void closeChannel(const std::string& id){
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channel_map.find(id);
            std::cout<<"当前信道个数:"<<_channel_map.size()<<std::endl;
            std::cout<<"剩余信道如下:"<<std::endl;
            for(auto& x:_channel_map){
                std::cout<<x.first<<std::endl;
            }
            if(it == _channel_map.end()){
               DLOG("要关闭的信道[%s]不存在",id.c_str());
               return ;
            }
            std::cout<<"开始删除信道"<<std::endl;
            _channel_map.erase(id);
        }
        Channel::ptr chooseChannel(const std::string& id){
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channel_map.find(id);
            if(it == _channel_map.end()){
               DLOG("指定的信道[%s]不存在",id.c_str());
               return Channel::ptr();
            }
            return it->second;
        }
    private:
      std::mutex _mutex;
      std::unordered_map<std::string , Channel::ptr> _channel_map;//用信道的id与信道对象建立映射
    };
}
#endif